package com.burges.net.tableAPIAndSQL.table

import java.lang

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.{Table, TableEnvironment, Types}
import org.apache.flink.table.catalog.InMemoryExternalCatalog
import org.apache.flink.table.sinks.CsvTableSink
import org.apache.flink.table.sources.CsvTableSource
import org.apache.flink.types.Row

/**
  * 创建人    BurgessLee
  * 创建时间   2020/2/13
  * 描述      创建环境实例代码
  */
object environment {

	def main(args: Array[String]): Unit = {
		//创建批量应用环境
		val batchEnv: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
		val tBathEnv = TableEnvironment.getTableEnvironment(batchEnv)

		// 创建实时应用
		val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
		val tStreamEnv = TableEnvironment.getTableEnvironment(streamEnv)

		/**
		  * 内部CataLog注册
		  */
		// 内部Table注册
//		val dataStream: DataStream[(Int, String)] = streamEnv.fromElements((192, "foo"), (122, "fun"))
//		tStreamEnv.registerTable("testRegisterInnerTable", dataStream)

		//TableSource注册
//		TableSource csvSource = new CsvTableSource("/path/to/file", "fileName")
//		tStreamEnv.registerTableSource("csvTable", csvSource)

		//TableSink注册
		// 指定csv文件
		val csvSink: CsvTableSink = new CsvTableSink("/path/csvfile/", ",")
		// 定义fieldNames和fieldTypes
		val fieldNames: Array[String] = Array("c1", "c2", "c3")
		val fieldTypes: Array[TypeInformation[_]] = Array(Types.INT, Types.DOUBLE, Types.LONG)
		// 将创建的csvSink注册到TableEnvironment中并指定名称为CsnSinkTable
		tStreamEnv.registerTableSink("csvSinkTable", fieldNames, fieldTypes, csvSink)

		/**
		  * 外部CataLog注册
		  */
		// 创建基于内存的外部CataLog
		val catalog: InMemoryExternalCatalog = new InMemoryExternalCatalog("")
		// 想TableEnvironment中注册创建好的InMEMCataLog
		tStreamEnv.registerExternalCatalog("inMemCataLog", catalog)

		/**
		  * DataStream/DataSet<=>Table
		  */
		//DataStream/DataSet=>Table注册方式
		val dataStream: DataStream[(Int, String)] = streamEnv.fromElements((192, "foo"), (122, "fun"))
		// 将DataStream注册成Table，指定表名为table1并使用默认字段名f0,f1
		tStreamEnv.registerDataStream("table1", dataStream)
		// 将DataStream注册成Table，指定表名为table1并使用默认字段名field1 field2
//		tStreamEnv.registerDataStream("table2", dataStream, 'fiedl1, `field2)

		//DataStream/DataSet=>Table转换方式
		// 将DataStream通过fromDataStream/fromDataSet转换成Table，不指定默认使用f0，f1作为字段名称
		val table1: Table = tStreamEnv.fromDataStream(dataStream)
		// 将DataStream通过fromDataStream/fromDataSet转换成Table，并指定字段名称
//		val table2: Table = tStreamEnv.fromDataStream(dataStream, 'field1, 'field2)

		//Table=>DataStream
		// 将table通过toAppendDataStream转换成Row格式的DataStream
		val dsRow: DataStream[Row] = tStreamEnv.toAppendStream[Row](table1)
		// 将table通过toAppendDataStream转换成Tuple2[Long,Int]格式的DataStream
		val dsTuple: DataStream[(Long, String)] = tStreamEnv.toAppendStream[(Long, String)](table1)
		// 将table通过toRetractStream转换成Row格式的DataStream
		// 根据第一个Boolean是否为True判断是插入还是删除引起的更新数据
		// True对应Insert操作更新的数据，false对应delete操作更新的数据
		val retractStream: DataStream[(Boolean, Row)] = tStreamEnv.toRetractStream[Row](table1)

		//Table=>DataSet
		val dataSet: DataSet[(Int, String)] = batchEnv.fromElements((192, "foo"), (122, "fun"))
		//从DataSet中创建Table
		val table3: Table = tBathEnv.fromDataSet(dataSet)
		//将Table转换成Row数据类型的DataSet数据集
		val rowDs: DataSet[Row] = tBathEnv.toDataSet[Row](table3)
		val tupleDataSet: DataSet[(Long, String)] = tBathEnv.toDataSet[(Long, String)](table3)

		/**
		  * Schema字段映射---字段位置映射
		  */
		// 字段名称默认使用默认值 _1 _2
		val tableSchema: Table = tStreamEnv.fromDataStream(dataStream)
		// 将DataStream注册成Table，指定表名为table1并使用默认字段名field1 field2
//		val tableSchema2: Table = tStreamEnv.fromDataStream(dataStream, 'field1, 'field2)

		/**
		  * Schema字段映射---字段名称映射
		  *  POJOS类型可以对属性同样可以重命名
		  */
		// 字段名称默认使用默认值 _1 _2
		val tableSchema3: Table = tStreamEnv.fromDataStream(dataStream)
		// 字段名称默认使用默认值 _2，并且仅输出字段名称是_2的字段
//		val tableSchema4: Table = tStreamEnv.fromDataStream(dataStream, '_2)
		// 将DataStream注册成Table，指定表名为table1并使用默认字段名field1 field2
//		val tableSchema5: Table = tStreamEnv.fromDataStream(dataStream, 'field1, 'field2)
		// 输出交换位置
//		val tableSchema6: Table = tStreamEnv.fromDataStream(dataStream, '_2, '_1)
		// 交换位置，并重命名
// 		val tableSchema7: Table = tStreamEnv.fromDataStream(dataStream, '_2 as 'field1, '_1 as 'field2)
	}

}
